﻿using log4net.Appender;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using log4net.Core;
using System.Threading;
using System.Threading.Tasks;
using System.Net;
using log4net.ElasticSearch.Model;

namespace log4net.ElasticSearch
{
    public class ElasticSearchAppender : BufferingAppenderSkeleton
    {
        static readonly string AppenderType = typeof(ElasticSearchAppender).Name;
        private object locker = new object();

        //Server=es.rfddc.com;Index=edi;Port=80;rolling=true
        private string m_host = "localhost";
        private int m_port = 9200;
        private string m_index = "log4net";
        private bool m_dateRooling = true;
        private int m_threadCount = 100;
        /// <summary>
        /// 默认一请求30000ms超时
        /// </summary>
        private int m_requestTimeout = 30000;
        /// <summary>
        /// 请求线程启动等待时间，默认为-1，表示永远等待，单位毫秒，
        /// 该属性可能会舍弃高并发下处理不过来的日志
        /// </summary>
        private int m_threadWaitTimeout = -1;
        /// <summary>
        /// 关闭时等待线程完成超时时间
        /// </summary>
        private int m_closeTimeout = 60000;
        /// <summary>
        /// 用于执行线程数限制
        /// </summary>
        private Semaphore m_semaphore;

        /// <summary>
        /// 当前请求数
        /// </summary>
        int RequestThreadCount;
        /// <summary>
        /// 总发起线程数据
        /// </summary>
        int TotalThreadCount;

        /// <summary>
        /// 用于关闭信号监测
        /// </summary>
        readonly ManualResetEvent workQueueEmptyEvent;

        #region 属性
        public String Host
        {
            get { return m_host; }
            set { m_host = value; }
        }
        public int Port
        {
            get { return m_port; }
            set { m_port = value; }
        }
        public String Index
        {
            get { return m_index; }
            set { m_index = value; }
        }
        public bool DateRooling
        {
            get { return m_dateRooling; }
            set { m_dateRooling = value; }
        }
        public int RequestTimeout
        {
            get { return m_requestTimeout; }
            set { m_requestTimeout = value; }
        }
        public int ThreadCount
        {
            set { m_threadCount = value; }
            get { return m_threadCount; }
        }
        public int CloseTimeout
        {
            get { return m_closeTimeout; }
            set { m_closeTimeout = value; }
        }
        public int ThreadWaitTimeout
        {
            get { return m_threadWaitTimeout; }
            set { m_threadWaitTimeout = value; }
        }

        Semaphore semaphore
        {
            get
            {
                if (m_semaphore == null)
                {
                    lock (locker)
                    {
                        if (m_semaphore == null)
                        {
                            m_semaphore = new Semaphore(m_threadCount, m_threadCount);
                        }
                    }
                }
                return m_semaphore;
            }
        }
        #endregion

        static ElasticSearchAppender()
        {
            //设置系统的最大连接数据
            if (ServicePointManager.DefaultConnectionLimit < 512)
            {
                ServicePointManager.DefaultConnectionLimit += 512;
            }
        }


        public ElasticSearchAppender()
        {
            workQueueEmptyEvent = new ManualResetEvent(true);
        }


        /// <summary>
        /// 实现接收消息类
        /// 整体异步执行
        /// </summary>
        /// <param name="events"></param>
        protected override void SendBuffer(LoggingEvent[] events)
        {
#if DEBUG
            var hash = this.GetHashCode();
#endif

            var logs = LogManager.ConvertToModel(events);
            var result = ThreadPool.QueueUserWorkItem(new WaitCallback(PostToES), logs);
            if (result == true)
            {
                workQueueEmptyEvent.Reset();
                Interlocked.Increment(ref TotalThreadCount);
            }
        }

        protected override void OnClose()
        {
            base.OnClose();
            if (!workQueueEmptyEvent.WaitOne(CloseTimeout, false))
            {
                var message = String.Format("关闭超时，可能未处理掉正在运行的队列，当前关闭超时时间为{0}毫秒，总线程数{1}，请求线程数据{2}",
                    CloseTimeout, TotalThreadCount, RequestThreadCount);
                HandleError(message);
            }
        }

        private void HandleError(string error, Exception ex = null)
        {
            var message = string.Format("{0} [{1}]: {2}.", AppenderType, Name, error);
            ErrorHandler.Error(message, ex);
        }

        /// <summary>
        /// 提交数据到ES
        /// </summary>
        /// <param name="obj"></param>
        private void PostToES(Object obj)
        {
            var isWaited = semaphore.WaitOne(ThreadWaitTimeout);
            try
            {
                if (isWaited == false)
                {
                    var message = String.Format("请求执行线程超，该数据将被舍弃，当前超时时间为{0}毫秒，总线程数{1}，请求线程数据{2}",
                        ThreadWaitTimeout, TotalThreadCount, RequestThreadCount);
                    throw new Exception(message);
                }
                Interlocked.Increment(ref RequestThreadCount);
                LogModel[] logModels = obj as LogModel[];
                if (logModels.Length > 1)
                {
                    //批量提交
                    PostBulk(logModels);
                }
                else
                {//单个提交
                    var logModel = logModels[0];
                    var jsonData = LogManager.ConvertToJson(logModel);

                    using (var wc = new EsWebClient(RequestTimeout))
                    {
                        var result = wc.UploadString(GetAddress(logModel.loggerName), jsonData);
                    }
                }

            }
            catch (Exception ex)
            {
                HandleError("发送数据失败（PostToES）", ex);
            }
            finally
            {
                if (isWaited)
                {
                    Interlocked.Decrement(ref RequestThreadCount);
                    semaphore.Release();
                }

                if (Interlocked.Decrement(ref TotalThreadCount) <= 0)
                {
                    workQueueEmptyEvent.Set();
                }
                //Console.Beep();
            }
        }

        /// <summary>
        /// 批量提交
        /// </summary>
        /// <param name="logModels"></param>
        private void PostBulk(LogModel[] logModels)
        {
            StringBuilder sb = new StringBuilder();
            foreach (var logModel in logModels)
            {
                sb.AppendLine("{ \"index\" : { \"_type\" : \"" + logModel.loggerName + "\" } }");
                sb.AppendLine(LogManager.ConvertToJson(logModel));
            }

            using (var wc = new EsWebClient(RequestTimeout))
            {
                var result = wc.UploadString(GetAddress(), sb.ToString());
            }
        }

        private Uri GetAddress(string type = null)
        {
            string index = m_index;
            if (m_dateRooling)
            {
                index = string.Format("{0}-{1}", m_index, DateTime.Now.ToString("yyyy.MM.dd"));
            }
            String url = String.Empty;
            if (String.IsNullOrWhiteSpace(type))
            {
                url = String.Format("http://{0}:{1}/{2}/_bulk", m_host, m_port, index.ToLower());
            }
            else
            {
                url = String.Format("http://{0}:{1}/{2}/{3}/", m_host, m_port, index.ToLower(), type);
            }
            return new Uri(url);
        }

    }
}
